package com.zst.kafka.nospring.client;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.text.MessageFormat;
import java.time.Duration;
import java.util.Properties;
import java.util.concurrent.locks.LockSupport;

public class ProducerRunner {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "192.168.23.204:19003");
        props.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.setProperty("linger.ms", String.valueOf(1));

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            while (true) {
                ProducerRecord<String, String> msg = new ProducerRecord<String, String>("test-topic", Long.toString(System.currentTimeMillis()),
                        MessageFormat.format("Current time millis = {0}", System.currentTimeMillis()));
                producer.send(msg, (recordMetadata, e) -> {
                    if (e != null) {
                        System.err.println("msg send err");
                        e.printStackTrace();
                    } else {
                        System.err.println("msg send success");
                    }
                });

                LockSupport.parkNanos(Duration.ofSeconds(2).toNanos());
            }
        }
    }
}
